package com.rj.data.sink;

import com.google.common.base.Charsets;
import com.google.common.base.Joiner;
import com.google.common.io.CharSink;
import com.google.common.io.FileWriteMode;
import com.google.common.io.Files;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class MapToFileSink implements Sink<Map<Integer, Integer[]>> {

    private String destFile;

    public MapToFileSink(String destPath) {
        this.destFile = destPath;
    }

    @Override
    public void writeLines(Map<Integer, Integer[]> maps, boolean append, boolean force, String splitter) throws IOException {
        if (!force && maps.size() < 100000) {
            return;
        }
        List<String> result = maps
                .entrySet()
                .stream()
                .map(et -> et.getKey() + "," + Joiner.on(",").join(et.getValue())).collect(Collectors.toList());

        CharSink sink = Files.asCharSink(new File(destFile), Charsets.UTF_8, FileWriteMode.APPEND);
        sink.writeLines(result);
        maps.clear();

    }
}
